package com.lagou.registry;

import com.lagou.server.NettyRpcServer;
import com.lagou.util.Constants;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

public class ZookeeperRegistryClient {
    Logger logger = LoggerFactory.getLogger(getClass().getName());


    //连接zookeeper客户端
    CuratorFramework zkClient;

    private String hostByservicePath;
    private NettyRpcServer nettyRpcServer;


    private ZookeeperRegistryClient() {
        //创建会话
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        zkClient = CuratorFrameworkFactory
                .builder()
                .namespace("base")
                .connectString("106.12.10.223:2181") //server地址
                .sessionTimeoutMs(5000) // 会话超时时间
                .connectionTimeoutMs(3000) // 连接超时时间
                .retryPolicy(retryPolicy) // 重试策略
                .build(); //
        zkClient.start();


    }

    static class Holder {
        static final ZookeeperRegistryClient instance = new ZookeeperRegistryClient();
    }

    public static ZookeeperRegistryClient getInstance() {
        return ZookeeperRegistryClient.Holder.instance;
    }


    public void register(RegistryInfo registryInfo) {
        try {
            String host = registryInfo.getHost();
            int port = registryInfo.getPort();

            //注册   服务名称/host:ip,  对应Host:ip路径的值设置该实现类名称
            String servicePath = Constants.ROOTPATH + registryInfo.getService();
            zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(servicePath + "/" + host + ":" + port, registryInfo.getServiceImpl().getBytes());

            //注册  hosts/host:/ip
            String path = zkClient.create().creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL).forPath(Constants.ROOTPATH + "hosts/" + host + ":" + port, (System.currentTimeMillis() + "-0").getBytes());


            PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, "/hosts", true);

            pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                    if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED)) {
                        System.out.println("添加子节点:" + event.getData().getPath());
                        if (!nettyRpcServer.isRunning) {
                            nettyRpcServer.start();
                        }

                    } else if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
                        logger.info("删除子节点:" + event.getData().getPath());
                        if (nettyRpcServer.isRunning) {
                            nettyRpcServer.stop();
                        }
                        String path = event.getData().getPath().replace("/hosts", "");
                        logger.info("====下线节点path：" + path);
                        //同时移除对应服务的节点
                        Stat stat = zkClient.checkExists().forPath(servicePath + path);

                        if (stat != null) {
                            zkClient.delete().forPath(servicePath + path);
                            //删除该节点后，判断该服务是否还有子节点，如果有就删除父类，代表没有节点注册该服务了

                            List<String> children = zkClient.getChildren().forPath(servicePath);
                            if (children == null || children.size() == 0) {
                                zkClient.delete().forPath(servicePath);
                            }
                        }


                    }

                }
            });

            pathChildrenCache.start(PathChildrenCache.StartMode.NORMAL);

            //每隔5s检查服务端响应时间是否有更新，没有就重置为0
            new ResponseThread("/hosts/" + host + ":" + port).start();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public Class getServiceByName(String servicename) {
        try {
            byte[] classNameBytes = zkClient.getData().forPath(hostByservicePath);
            String className = new String(classNameBytes, "utf-8");
            Class serviceImplClass = Class.forName(className);

            return serviceImplClass;
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("未找到对应的值：" + hostByservicePath);
        }

        return null;
    }


    public void setHostByservicePath(String hostByservicePath) {
        this.hostByservicePath = hostByservicePath;
    }

    public void setServer(NettyRpcServer nettyRpcServer) {
        this.nettyRpcServer = nettyRpcServer;
    }


    class ResponseThread extends Thread {

        private String path;

        public ResponseThread(String path) {
            this.path = path;
        }

        @Override
        public void run() {
            while (true) {
                try {
                    Thread.sleep(6000);

                    logger.info("检查响应时间：{}", path);
                    byte[] value = zkClient.getData().forPath(path);


                    String times = new String(value);
                    long lastTime = Long.parseLong(times.split("-")[0]);
                    long responseTime = Long.parseLong(times.split("-")[1]);
                    if (System.currentTimeMillis() - lastTime > 5000) {
                        if (responseTime != 0) {
                            zkClient.setData().forPath(path, (System.currentTimeMillis() + "-0").getBytes());
                        }
                    }


                } catch (Exception e) {
                    e.printStackTrace();
                }


            }
        }
    }
}
